package org.example.ItemCF.src.recommend;

import org.apache.hadoop.io.IntWritable;
import org.apache.hadoop.io.LongWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Mapper;

import java.io.IOException;
import java.util.ArrayList;
import java.util.HashMap;
import java.util.List;
import java.util.Map;

/**
 * 此Mapper类用于执行物品局部乘法，将用户对物品的偏好与物品之间的共同出现次数相乘，
 * 生成中间键值对，其中键为用户ID，值为与该用户相关的所有物品及其调整后的偏好值。
 */
public class Step4PartialMultiplyMapper extends
        Mapper<LongWritable, Text, IntWritable, Text> {
    private final static IntWritable k = new IntWritable();
    private final static Text v = new Text();

    // 用于存储物品对和它们的共同出现次数的映射
    private final static Map<Integer, List<Cooccurrence>> cooccurrenceMatrix = new HashMap<Integer, List<Cooccurrence>>();

    /**
     * 根据输入的键值对（行号，值）进行处理，将物品对和它们的共同出现次数存储到cooccurrenceMatrix中，
     * 或者将用户ID和调整后的物品偏好值写入输出上下文中。
     */
    @Override
    public void map(LongWritable key, Text values, Context context)
            throws IOException, InterruptedException {
        String[] tokens = Recommend.DELIMITER.split(values.toString());

        String[] v1 = tokens[0].split(":");
        String[] v2 = tokens[1].split(":");

        // 处理物品共同出现次数的数据
        if (v1.length > 1) {
            int itemID1 = Integer.parseInt(v1[0]);
            int itemID2 = Integer.parseInt(v1[1]);
            int num = Integer.parseInt(tokens[1]);

            List<Cooccurrence> list = null;
            if (!cooccurrenceMatrix.containsKey(itemID1)) {
                list = new ArrayList<Cooccurrence>();
            } else {
                list = cooccurrenceMatrix.get(itemID1);
            }
            list.add(new Cooccurrence(itemID1, itemID2, num));
            cooccurrenceMatrix.put(itemID1, list);
        }

        // 处理用户偏好数据
        if (v2.length > 1) {
            int itemID = Integer.parseInt(tokens[0]);
            int userID = Integer.parseInt(v2[0]);
            double pref = Double.parseDouble(v2[1]);
            k.set(userID);
            for (Cooccurrence co : cooccurrenceMatrix.get(itemID)) {
                v.set(co.getItemID2() + "," + pref * co.getNum());
                context.write(k, v);
            }
        }
    }
}
   
